//package cn.com.niit.socketio;
//
//import com.corundumstudio.socketio.SocketIOClient;
//import org.apache.kafka.clients.consumer.ConsumerConfig;
//import org.apache.kafka.clients.consumer.ConsumerRecord;
//import org.apache.kafka.clients.consumer.ConsumerRecords;
//import org.apache.kafka.clients.consumer.KafkaConsumer;
//import java.time.Duration;
//import java.util.Arrays;
//import java.util.Properties;
//
//public class SocketThread extends Thread {
//    private SocketIOClient client;
//    private boolean flag;
//
//    public SocketThread(SocketIOClient client) {
//        this.client = client;
//        flag = true;
//    }
//
//    @Override
//    public void run() {
//        Properties props = new Properties();
//        props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.136.128:9092");
//        props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "niit");
//        props.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
//        props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
//        props.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
//        props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
//        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(props);
//        // 订阅 5 个需要的主题
//        kafkaConsumer.subscribe(Arrays.asList("classStu", "semesterclassStu", "semeterSex", "stateStu", "stuSexcount"));
//
//        while (flag) {
//            ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(5));
//            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
//                String topic = consumerRecord.topic();
//                String value = consumerRecord.value();
//                System.out.println("Received data from topic " + topic + ": " + value);
//                if (client!= null && client.isChannelOpen()) {
//                    // 根据不同主题发送不同格式消息
//                    //张弈婷
//                    if ("classStu".equals(topic)) {
//                        // 假设数据格式为 class_id,gender,count
//                        String[] dataArray = value.split(",");
//                        String class_id = dataArray[0];
//                        String gender = dataArray[1];
//                        int count = Integer.parseInt(dataArray[2]);
//                        StringBuilder message = new StringBuilder();
//                        message.append("classmsg:").append(class_id).append(",").append(gender).append(",").append(count);
//                        client.sendEvent("classmsg", message.toString());
//                    } else if ("semesterclassStu".equals(topic)) {
//                        // 假设数据格式为 class_id,semester,gender,count
//                        //李一鸣
//                        String[] dataArray = value.split(",");
//                        String class_id = dataArray[0];
//                        String semester = dataArray[1];
//                        String gender = dataArray[2];
//                        int count = Integer.parseInt(dataArray[3]);
//                        StringBuilder message = new StringBuilder();
//                        message.append("semesterclassmsg:").append(class_id).append(",").append(semester).append(",").append(gender).append(",").append(count);
//                        client.sendEvent("semesterclassmsg", message.toString());
//                    } else if ("semeterSex".equals(topic)) {
//                        // 假设数据格式为 semeter,gender,count
//                        //孟佳怡
//                        String[] dataArray = value.split(",");
//                        String semeter = dataArray[0];
//                        String gender = dataArray[1];
//                        int count = Integer.parseInt(dataArray[2]);
//                        StringBuilder message = new StringBuilder();
//                        message.append("semetermsg:").append(semeter).append(",").append(gender).append(",").append(count);
//                        client.sendEvent("semetermsg", message.toString());
//                    } else if ("stateStu".equals(topic)) {
//                        // 假设数据格式为 state,gender,count
//                        //杨茗茹
//                        String[] dataArray = value.split(",");
//                        String state = dataArray[0];
//                        String gender = dataArray[1];
//                        int count = Integer.parseInt(dataArray[2]);
//                        StringBuilder message = new StringBuilder();
//                        message.append("statemsg:").append(state).append(",").append(gender).append(",").append(count);
//                        client.sendEvent("statemsg", message.toString());
//                    } else if ("stuSexcount".equals(topic)) {
//                        // 假设数据格式为 gender,count
//                        //柳炎炎
//                        String[] dataArray = value.split(":");
//                        String gender = dataArray[0];
//                        int count = Integer.parseInt(dataArray[1]);
//                        StringBuilder message = new StringBuilder();
//                        message.append("stumsg:").append(gender).append(",").append(count);
//                        client.sendEvent("stumsg", message.toString());
//                    }
//                    System.out.println("Data sent successfully");
//                } else {
//                    flag = false;
//                }
//            }
//        }
//        kafkaConsumer.close();
//        System.out.println("线程结束...");
//    }
//}
package cn.com.niit.socketio;

import com.corundumstudio.socketio.SocketIOClient;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.time.Duration;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

public class SocketThread extends Thread {
    private SocketIOClient client;
    private boolean flag;
    // 用于存储学期性别统计数据 孟佳怡
    private Map<String, Map<String, Integer>> semesterGenderCount = new HashMap<>();
    // 用于存储性别统计数据  柳炎炎
    private Map<String, Integer> genderCount = new HashMap<>();
    // 用于存储状态性别统计数据  杨茗茹
    private Map<String, Map<String, Integer>> stateGenderCount = new HashMap<>();
    // 用于存储班级性别统计数据  张羿婷
    private Map<String, Map<String, Integer>> classGenderCount = new HashMap<>();
    // 用于存储学期班级性别统计数据  李一鸣
    private Map<String, Map<String, Map<String, Integer>>> semesterClassGenderCount = new HashMap<>();

    public SocketThread(SocketIOClient client) {
        this.client = client;
        flag = true;
    }

    @Override
    public void run() {
        Properties props = new Properties();
        props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.136.128:9092");//改成自己的ip
        props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "niit");
        props.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
        props.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(props);
        // 订阅 stuInfo 这一个主题
        kafkaConsumer.subscribe(Arrays.asList("stuInfo"));

        while (flag) {
            ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(5));
            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                // 每次处理新数据前，先清空之前的统计数据
                resetStatistics();

                String value = consumerRecord.value();
                System.out.println("Received data from stuInfo topic: " + value);
                if (client!= null && client.isChannelOpen()) {
                    // 按照制表符分割数据
                    String[] fields = value.split("\t");
                    // 获取相关字段，所有的功能的都用这个
                    String class_id = fields[0];
                    int genderInt = Integer.parseInt(fields[2]);
                    String gender = genderInt == 0? "female" : "male";
                    String semester = fields[5];
                    String status = fields[6];

                    // 班级性别统计 张羿婷
                    classGenderCount.computeIfAbsent(class_id, k -> new HashMap<>()).compute(gender, (k1, v) -> (v == null)? 1 : v + 1);
                    // 学期班级性别统计  李一鸣
                    semesterClassGenderCount.computeIfAbsent(semester, k -> new HashMap<>()).computeIfAbsent(class_id, k1 -> new HashMap<>()).compute(gender, (k2, v) -> (v == null)? 1 : v + 1);
                    // 学期性别统计 孟佳怡
                    semesterGenderCount.computeIfAbsent(semester, k -> new HashMap<>()).compute(gender, (k1, v) -> (v == null)? 1 : v + 1);
                    // 状态性别统计  杨茗茹
                    stateGenderCount.computeIfAbsent(status, k -> new HashMap<>()).compute(gender, (k1, v) -> (v == null)? 1 : v + 1);
                    // 性别统计 柳炎炎
                    genderCount.compute(gender, (k, v) -> (v == null)? 1 : v + 1);

                    // 发送班级性别统计结果 张羿婷
                    for (Map.Entry<String, Map<String, Integer>> classEntry : classGenderCount.entrySet()) {
                        String classKey = classEntry.getKey();
                        for (Map.Entry<String, Integer> genderEntry : classEntry.getValue().entrySet()) {
                            String genderKey = genderEntry.getKey();
                            int count = genderEntry.getValue();
                            StringBuilder classMsg = new StringBuilder();
                            classMsg.append("classmsg:").append(classKey).append(",").append(genderKey).append(",").append(count);
                            client.sendEvent("classmsg", classMsg.toString());
                            System.out.println("Processed classmsg: " + classMsg.toString());
                        }
                    }
                    // 发送学期班级性别统计结果  李一鸣
                    for (Map.Entry<String, Map<String, Map<String, Integer>>> semesterEntry : semesterClassGenderCount.entrySet()) {
                        String semesterKey = semesterEntry.getKey();
                        for (Map.Entry<String, Map<String, Integer>> classEntry : semesterEntry.getValue().entrySet()) {
                            String classKey = classEntry.getKey();
                            for (Map.Entry<String, Integer> genderEntry : classEntry.getValue().entrySet()) {
                                String genderKey = genderEntry.getKey();
                                int count = genderEntry.getValue();
                                StringBuilder semesterClassMsg = new StringBuilder();
                                semesterClassMsg.append("semesterclassmsg:").append(classKey).append(",").append(semesterKey).append(",").append(genderKey).append(",").append(count);
                                client.sendEvent("semesterclassmsg", semesterClassMsg.toString());
                                System.out.println("Processed semesterclassmsg: " + semesterClassMsg.toString());
                            }
                        }
                    }
                    // 发送学期性别统计结果 孟佳怡
                    for (Map.Entry<String, Map<String, Integer>> semesterEntry : semesterGenderCount.entrySet()) {
                        String semesterKey = semesterEntry.getKey();
                        for (Map.Entry<String, Integer> genderEntry : semesterEntry.getValue().entrySet()) {
                            String genderKey = genderEntry.getKey();
                            int count = genderEntry.getValue();
                            StringBuilder semesterMsg = new StringBuilder();
                            semesterMsg.append("semetermsg:").append(semesterKey).append(",").append(genderKey).append(",").append(count);
                            client.sendEvent("semetermsg", semesterMsg.toString());
                            System.out.println("Processed semetermsg: " + semesterMsg.toString());
                        }
                    }
                    // 发送状态性别统计结果 杨茗茹
                    for (Map.Entry<String, Map<String, Integer>> stateEntry : stateGenderCount.entrySet()) {
                        String stateKey = stateEntry.getKey();
                        for (Map.Entry<String, Integer> genderEntry : stateEntry.getValue().entrySet()) {
                            String genderKey = genderEntry.getKey();
                            int count = genderEntry.getValue();
                            StringBuilder stateMsg = new StringBuilder();
                            stateMsg.append("statemsg:").append(stateKey).append(",").append(genderKey).append(",").append(count);
                            client.sendEvent("statemsg", stateMsg.toString());
                            System.out.println("Processed statemsg: " + stateMsg.toString());
                        }
                    }
                    // 发送性别统计结果 柳炎炎
                    for (Map.Entry<String, Integer> genderEntry : genderCount.entrySet()) {
                        String genderKey = genderEntry.getKey();
                        int count = genderEntry.getValue();
                        StringBuilder stuMsg = new StringBuilder();
                        stuMsg.append("stumsg:").append(genderKey).append(",").append(count);
                        client.sendEvent("stumsg", stuMsg.toString());
                        System.out.println("Processed stumsg: " + stuMsg.toString());
                    }

                    System.out.println("Data sent successfully");
                } else {
                    flag = false;
                }
            }
        }
        kafkaConsumer.close();
        System.out.println("线程结束...");
    }

    private void resetStatistics() {
        semesterGenderCount.clear();
        genderCount.clear();
        stateGenderCount.clear();
        classGenderCount.clear();
        semesterClassGenderCount.clear();
    }
}